package com.roc.rocketmqconsumer.messageType;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 普通消息-消费者
 *
 * @author Roker
 * @date 2025/07/06
 */
public class NormalMsgConsumer {
    public static void main(String[] args) throws MQClientException {
        // 实例化消费者，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("normalMsgConsumerGroup");
        // 指定NameServer地址,需与生产者端保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 订阅一个或多个 Topic，以及对应的 Tag 来过滤消息，* 表示订阅该 Topic 下所有 Tag
        consumer.subscribe("TOPIC_NORMAL", "*");
        // 注册消息监听器(回调实现类),处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        // 输出消息内容，这里简单打印消息体（需注意编码，示例假设是 UTF-8）
                        String msgBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("接收消息：topic=" + msg.getTopic() + ", tag=" + msg.getTags() + ", 消息内容=" + msgBody);
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 若消息处理异常，可根据情况决定是否重试等，这里返回 RECONSUME_LATER 表示稍后重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 消息处理成功，返回 CONSUME_SUCCESS
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者实例
        consumer.start();
        System.err.println("消费者已启动，等待接收消息...");


    }
}
